package com.study.flink.scala.day02_xcy

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

class KafkaMessageWatermarksScala extends AssignerWithPeriodicWatermarks[String]{
  private var currentMaxTimestamp = 0L
  private var currentTimestamp:Long = Long.MinValue

  //获取当前系统时间，要做延迟在这里做，比如延迟5秒，就把返回时间扣去5秒
  override def getCurrentWatermark: Watermark = {
//    var time:Long = Long.MinValue
//    if(currentTimestamp == Long.MinValue) {
//      time = Long.MinValue
//    }else {
//      time = this.currentTimestamp - 1
//    }
//
//    new Watermark(time)
    new Watermark(currentMaxTimestamp-System.currentTimeMillis())
//    new Watermark(System.currentTimeMillis())
  }

  //从数据中获取EventTIme的逻辑
  override def extractTimestamp(element: String, previousElementTimestamp: Long): Long = {
    this.currentTimestamp = System.currentTimeMillis()
    this.currentTimestamp
  }
}
